package stress

import (
	"time"

	"mongo-stress/point"
	"mongo-stress/write"
)

// WriteResult contains the latency, status code, and error type
// each time a write happens.
type WriteResult struct {
	LatNs      int64
	StatusCode int
	Body       string // Only populated when unusual status code encountered.
	Err        error
	Timestamp  int64
}

// WriteConfig specifies the configuration for the Write function.
type WriteConfig struct {
	BatchSize uint64
	MaxPoints uint64
	Interval  time.Duration
	BaseTimes []time.Time
	Counts    []uint64
	Offset    int
	Deadline  time.Time
	Tick      <-chan time.Time
}

// Write takes in a slice of lineprotocol.Points, a write.Client, and a WriteConfig. It will attempt
// to write data to the target until one of the following conditions is met.
// 1. We reach that MaxPoints specified in the WriteConfig.
// 2. We've passed the Deadline specified in the WriteConfig.
func Write(pts []*point.Point, c write.Client, cfg WriteConfig) (uint64, time.Duration) {

	var pointCount uint64
	var docs []interface{}

	start := time.Now()
	// buf := bytes.NewBuffer(nil)
	t := time.Now()

	// var w io.Writer = buf
	tPrev := t
WRITE_BATCHES:
	for {
		if t.After(cfg.Deadline) {
			break WRITE_BATCHES
		}

		if pointCount >= cfg.MaxPoints {
			break
		}

		for i, pt := range pts {
			pointCount++
			// 计算当前点的时间
			ts := cfg.BaseTimes[cfg.Offset+i].Add(time.Duration(cfg.Counts[cfg.Offset+i]) * cfg.Interval)

			cfg.Counts[cfg.Offset+i]++
			newFields := pt.CloneFields()
			pt.Update(newFields)
			docs = append(docs, map[string]interface{}{
				"tags":      pt.Tags,
				"fields":    newFields,
				"timestamp": ts.UTC(),
			})
			//fmt.Printf("Insert doc - tags: %v, timestamp: %s\n", pt.Tags, ts.UTC().Format(time.RFC3339Nano))
			if pointCount%cfg.BatchSize == 0 {
				sendBatch(c, docs)
				docs = docs[:0] // 避免重复写入
				t = <-cfg.Tick
				if t.After(cfg.Deadline) {
					break WRITE_BATCHES
				}

				if pointCount >= cfg.MaxPoints {
					break
				}

			}

		}

		// Avoid timestamp colision when batch size > pts
		if t.After(tPrev) {
			tPrev = t
			continue
		}
		t = t.Add(1 * time.Nanosecond)
	}

	return pointCount, time.Since(start)
}

func sendBatch(c write.Client, docs []interface{}) {
	// pointsWritten, lat, status, body, err := c.Send(docs)
	// select {
	// case ch <- WriteResult{LatNs: lat, StatusCode: status, Body: body, Err: err, Timestamp: time.Now().UnixNano()}:
	// default:
	// }
	c.Send(docs)
}
